package POE::Filter::Gearman;

use strict;
use warnings;
use Smart::Comments;

use POE::Filter;
use POE::Component::Gearman::Constants;

our @ISA = qw(POE::Filter);

use Carp qw(croak);

sub BUFFER      () { 0 } # raw data buffer to build requests
sub STATE       () { 1 } # built a full request
sub PACKET      () { 2 } # built a full request

sub ST_HEADERS  () { 0x01 } # waiting for complete header block
sub ST_CONTENT  () { 0x02 } # waiting for complete body

sub HEADER_LEN  () { 12 }   # header length

sub new {
    my $class = shift;
    return bless (
        [
            '',         # BUFFER
            ST_HEADERS, # STATE
            {},         # PACKET
        ],
        $class
    );
}

sub get_one_start {
    my ($self, $stream) = @_;
    $self->[BUFFER] .= join( '', @$stream );
}

sub get_one {
    my ($self, $stream) = @_;

    if ( $self->[STATE] == ST_HEADERS ){
        if ( length $self->[BUFFER] < HEADER_LEN ){
            return [];
        }

        my $magic = substr( $self->[BUFFER], 0, 1 );
        if ( $magic eq "\0" ){   # Binary Protocol
            my $header = substr $self->[BUFFER], 0, HEADER_LEN, '';
            my ($way, $code, $len) = unpack 'x1 a3 N N', $header;

            $self->[PACKET] = {
                way     => $way,
                code    => POE::Component::Gearman::Constants::code2name($code),
                len     => $len,
            };

            $self->[STATE] = ST_CONTENT;
        }
        else {                                          # Line Protocol
            return [];
        }
    }

    if ( $self->[STATE] == ST_CONTENT ){
        if ( length $self->[BUFFER] < $self->[PACKET]->{len} ){
            return [];
        }

        my $data = substr $self->[BUFFER], 0, $self->[PACKET]->{len}, '';
        my @data = split "\x00", $data;

        $self->[PACKET]->{data} = \@data;
        my $pkt = $self->[PACKET];
        delete $pkt->{len};

        $self->[STATE] = ST_HEADERS;
        $self->[PACKET] = {};

        return [ $pkt ];
    }
}

sub put {
    my ($self, $pkts) = @_;
    my @raw;

    foreach my $pkt (@$pkts) {
        $pkt->{data} = '' if not defined $pkt->{data};
        my $data = !ref $pkt->{data}
                        ? ($pkt->{data})
                        : join "\x00", map {
                                defined $_ ? $_ : ''
                            } @{ $pkt->{data} };
        $pkt->{code} = POE::Component::Gearman::Constants::name2code($pkt->{code}),
        my $str = pack 'a1 a3 N N/a*', "\x00", $pkt->{way}, $pkt->{code}, $data;
        push @raw, $str;
    }

    \@raw;
}

sub get_pending {
  my $self = shift;
  croak ref($self)." does not support the get_pending() method\n";
  return;
}

1;
